package data.FlinkJoin

import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}

import scala.collection.mutable.ListBuffer

/**
 * 名字,城市,时间戳
 **/
class BSource extends RichParallelSourceFunction[(String,String,Long)]{
  var running = true

  override def run(ctx: SourceFunction.SourceContext[(String, String, Long)]): Unit = {
    val buffer = ListBuffer[(String, String, Long)]()
    buffer.+=(("PK","北京",50000L)) // [50,60)
    buffer.+=(("RZ","深圳",109000L)) // [100,110)

    var count = 0
    while(running && count < buffer.length) {
      ctx.collect((buffer(count)._1,buffer(count)._2,buffer(count)._3))
      count += 1

      Thread.sleep(1000)
    }

  }

  override def cancel(): Unit = {

    running = false
  }
}
